import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class Demo4Filter {
    Configuration conf = null;
    Connection conn = null;
    Table students = null;

    public void printRS(ResultScanner scanner) throws IOException {
        Result rs = scanner.next();
        while (rs != null) {
            List<Cell> cells = rs.listCells();
            String resultStr = "";
            String rk = Bytes.toString(rs.getRow());
            resultStr = resultStr + rk;
            for (Cell cell : cells) {
                String value = Bytes.toString(rs.getValue(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell)));
                resultStr = resultStr + "\t" + value;
            }
            System.out.println(resultStr);
            rs = scanner.next();

        }
    }

    @Before
    public void init() {
        conf = HBaseConfiguration.create();
        // 其他配置参考http://hbase.apache.org/1.4/book.html#config.files
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
            // 获取students表
            students = conn.getTable(TableName.valueOf("students"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    // 通过RowFilter过滤比rowKey 1500100010 小的所有值出来
    public void BinaryComparatorFilter() throws IOException {
        // 传入需要比较的值，创建BinaryComparator二进制比较器
        BinaryComparator binaryComparator = new BinaryComparator("1500100010".getBytes());
        // 传入比较的操作、比较器 创建RowFilter过滤器
        RowFilter rowFilter = new RowFilter(CompareOp.LESS, binaryComparator);

        Scan scan = new Scan();
        // 设置过滤器
        scan.setFilter(rowFilter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);
    }

    @Test
    // 通过FamilyFilter查询列簇名包含nf的所有列簇下面的数据
    public void SubstringComparatorFilter() throws IOException {
        SubstringComparator substringComparator = new SubstringComparator("f");
        FamilyFilter familyFilter = new FamilyFilter(CompareOp.EQUAL, substringComparator);

        Scan scan = new Scan();
        scan.setFilter(familyFilter);

        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);

    }

    // 将列名包含 a 的所有列的值取出来
    @Test
    public void SubstringComparatorQualifierFilter() throws IOException {
        SubstringComparator substringComparator = new SubstringComparator("a");
        QualifierFilter qualifierFilter = new QualifierFilter(CompareOp.EQUAL, substringComparator);

        Scan scan = new Scan();
        scan.setFilter(qualifierFilter);

        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);

    }

    // 把列值以文科开头的cell全部过滤出来
    @Test
    public void BinaryPrefixComparatorFilter() throws IOException {
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("文科".getBytes());
        ValueFilter valueFilter = new ValueFilter(CompareOp.NOT_EQUAL, binaryPrefixComparator);
        Scan scan = new Scan();
        scan.setFilter(valueFilter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);

    }

    // 过滤出rowkey包含8的学生信息
    @Test
    public void RegexStringComparatorFilter() throws IOException {
        RegexStringComparator regexStringComparator = new RegexStringComparator(".*8.*");
        RowFilter rowFilter = new RowFilter(CompareOp.EQUAL, regexStringComparator);
        Scan scan = new Scan();
        scan.setFilter(rowFilter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);
    }

    @Test
    public void TestBitComparator() throws IOException {
        BitComparator bitComparator = new BitComparator(Bytes.toBytes("1500100014"), BitComparator.BitwiseOp.OR);
        RowFilter rowFilter = new RowFilter(CompareOp.EQUAL, bitComparator);
        Scan scan = new Scan();
        scan.setFilter(rowFilter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);
    }

    // age > 22
    @Test
    public void SingleColumnValueFilterTest() throws IOException {
        SingleColumnValueFilter filter = new SingleColumnValueFilter(
                "info".getBytes(),
                "age".getBytes(),
                CompareOp.GREATER,
                "22".getBytes()
        );
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);

    }

    // 过滤出文科的学生
    @Test
    public void SingleColumnValueFilterTest2() throws IOException {
        SingleColumnValueFilter filter = new SingleColumnValueFilter(
                "info".getBytes(),
                "clazz".getBytes(),
                CompareOp.EQUAL,
                new BinaryPrefixComparator("文科".getBytes())
        );
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);

    }

    // 过滤出文科的学生 不返回比较的列
    @Test
    public void SingleColumnValueExcludeFilterTest2() throws IOException {
        SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter(
                "info".getBytes(),
                "clazz".getBytes(),
                CompareOp.EQUAL,
                new BinaryPrefixComparator("文科".getBytes())
        );
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);
    }

    // 通过PrefixFilter查询以150010008开头的所有前缀的rowkey
    @Test
    public void PrefixFilterTest() throws IOException {
        PrefixFilter prefixFilter = new PrefixFilter("150010008".getBytes());
        Scan scan = new Scan();
        scan.setFilter(prefixFilter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);
    }

    public void printScanWithFilter(Filter filter) throws IOException {
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);
    }

    @Test
    public void PageFilterTest() throws IOException {
        int page = 5;
        int pageSize = 10;
        int current_page_first_num = (page - 1) * pageSize + 1;
        PageFilter pageFilter = new PageFilter(current_page_first_num);
        Scan scan = new Scan();
        scan.setFilter(pageFilter);
        ResultScanner scanner = students.getScanner(scan);
        byte[] current_page_first_rowkey = null;
        for (Result rs : scanner) {
            current_page_first_rowkey = rs.getRow();
        }
        scan.withStartRow(current_page_first_rowkey);
        PageFilter pageFilter1 = new PageFilter(pageSize);
        scan.setFilter(pageFilter1);
        ResultScanner scanner1 = students.getScanner(scan);
        printRS(scanner1);
    }

    @Test
    // 通过合理的设置rowkey来实现分页功能，提高效率
    public void PageFilterTest2() throws IOException {
        int PageSize = 10;
        int PageNum = 3;

        int baseId = 1500100000;
        int start_row = baseId + (PageNum - 1) * PageSize + 1;
        int end_row = start_row + PageSize;
        Scan scan = new Scan();
        scan.withStartRow(String.valueOf(start_row).getBytes());
        scan.withStopRow(String.valueOf(end_row).getBytes());
        ResultScanner scanner = students.getScanner(scan);
        printRS(scanner);
    }

    // rowkey包含8，age>22,文科班的学生信息
    @Test
    public void FilterListFilter() throws IOException {
        SubstringComparator substringComparator = new SubstringComparator("8");
        RowFilter rowFilter = new RowFilter(CompareOp.EQUAL, substringComparator);

        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(
                "info".getBytes(),
                "age".getBytes(),
                CompareOp.GREATER,
                "22".getBytes()
        );


        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(
                "info".getBytes(),
                "clazz".getBytes(),
                CompareOp.EQUAL,
                new BinaryPrefixComparator("文科".getBytes())
        );

        FilterList filterList = new FilterList();
        filterList.addFilter(rowFilter);
        filterList.addFilter(filter1);
        filterList.addFilter(filter2);

        printScanWithFilter(filterList);

    }

    @After
    public void closeAll() throws IOException {
        if (conn != null) {
            conn.close();
        }
    }
}
